package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author:ray
 * @time:2022/10/15 10:15
 * @description:
 **/
public class DwdTradeOrderDetail {
    public static void main(String[] args) {
        //TODO 1、环境配置
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2、检查点设置
        //2.1 设置检查点
        /*env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点的超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 job 取消之后检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");*/

        //TODO 3、从kafka的订单预处理主题中读取数据创建动态表
        tableEnv.executeSql(
                "create table dwd_trade_order_pre_process(" +
                " `type` string,\n" +
                " oi_ts string,\n" +
                " `old` map<string,string>,\n" +
                " user_id string,\n" +
                " order_status string,\n" +
                " session_id string,\n" +
                " province_id string,\n" +
                " operate_date string,\n" +
                " operate_time string,\n" +
                " id string,\n" +
                " course_id string,\n" +
                " course_name string,\n" +
                " order_id string,\n" +
                " origin_amount string,\n" +
                " coupon_reduce string,\n" +
                " final_amount string,\n" +
                " create_date string,\n" +
                " create_time string,\n" +
                " od_ts string,\n" +
                " row_op_ts timestamp_ltz(3)\n"+
                ")"  + MyKafkaUtil.getKafkaDDL(
            "dwd_trade_order_pre_process", "dwd_trade_order_detail"));


        //TODO 4、过滤出下单数据
        Table dwdOrderDetail = tableEnv.sqlQuery("select \n" +
                " id,\n" +
                " user_id,\n" +
                " session_id,\n" +
                " province_id,\n" +
                " course_id,\n" +
                " course_name,\n" +
                " order_id,\n" +
                " origin_amount,\n" +
                " coupon_reduce,\n" +
                " final_amount,\n" +
                " create_date,\n" +
                " create_time,\n" +
                " od_ts,\n" +
                " row_op_ts\n" +
                "from dwd_trade_order_pre_process\n" +
                "where `type` = 'insert'"
        );

        //TODO 5、创建 Kafka 下单明细表
        tableEnv.executeSql("create table dwd_order_detail(\n" +
                " id string,\t\n" +
                " user_id string,\n" +
                " session_id string, \n" +
                " province_id string,\n" +
                " course_id string,\n" +
                " course_name string,\n" +
                " order_id string,\n" +
                " origin_amount string,\n" +
                " coupon_reduce string,\n" +
                " final_amount string,\n" +
                " create_date string,\n" +
                " create_time string,\n" +
                " od_ts string,\n" +
                " row_op_ts timestamp_ltz(3),\n" +
                " primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_order_detail")) ;

        //TODO 6、将数据写出到 Kafka
        dwdOrderDetail.executeInsert("dwd_order_detail");

    }
}
